package cn.xiaobai.spring.boot.config;

import cn.xiaobai.spring.boot.domain.KafkaPlusProperties;
import cn.xiaobai.spring.boot.listener.BatchKafkaListener;
import cn.xiaobai.spring.boot.listener.KafkaPlusListenerAdapter;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.Map;

/**
 * kafka 自动配置
 *
 * @Author yangdaji
 * @Date 2023/11/4
 * @PackageName cn.xiaobai.config
 * @ClassName KafkaAutoConfiguration
 */
@AutoConfiguration
@EnableConfigurationProperties({KafkaPlusProperties.class})
@Import({BatchKafkaListener.class})
public class KafkaAutoConfiguration {

    private final KafkaPlusProperties kafkaPlusProperties;

    public KafkaAutoConfiguration(KafkaPlusProperties kafkaPlusProperties) {
        this.kafkaPlusProperties = kafkaPlusProperties;
    }

    @Bean
    public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> defaultKafkaProducerFactory) {
        return new KafkaTemplate<>(defaultKafkaProducerFactory);
    }

    /**
     * 消息监听器容器
     *
     * @return
     */
    @Bean
    public ConcurrentMessageListenerContainer<?, ?> messageListenerContainer(ConcurrentKafkaListenerContainerFactory<Object, Object> concurrentKafkaListenerContainerFactory
            , BatchKafkaListener batchKafkaListener) {
        ConcurrentMessageListenerContainer<Object, Object> listenerContainer =
                concurrentKafkaListenerContainerFactory.createListenerContainer(new KafkaPlusListenerAdapter(batchKafkaListener));
        listenerContainer.setupMessageListener(batchKafkaListener);
        return listenerContainer;
    }


    /**
     * 消费者工厂
     *
     * @return
     */
    @Bean
    public ConsumerFactory<?, ?> kafkaConsumerFactory() {
        Map<String, Object> consumerProperties = kafkaPlusProperties.buildConsumerProperties();
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    /**
     * 生产者工厂
     *
     * @return
     */
    @Bean
    public ProducerFactory<?, ?> kafkaProducerFactory() {
        Map<String, Object> producerProperties = kafkaPlusProperties.buildProducerProperties();
        return new DefaultKafkaProducerFactory<>(producerProperties);
    }

    /**
     * 消息监听器容器工厂
     *
     * @param consumerFactory
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> concurrentKafkaListenerContainerFactory =
                new ConcurrentKafkaListenerContainerFactory<>();
        concurrentKafkaListenerContainerFactory.setBatchListener(true);
        concurrentKafkaListenerContainerFactory.setConsumerFactory(consumerFactory);
        return concurrentKafkaListenerContainerFactory;
    }


    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("newTopic").build();
    }

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> properties = kafkaPlusProperties.buildConsumerProperties();
        return new KafkaAdmin(properties);
    }

}
